//! Redis风格的连接处理器
//! 
//! 基于Redis官方实现思路的优化方案：
//! 1. 单线程事件循环处理所有连接
//! 2. 每个连接独立的输入输出缓冲区
//! 3. 在同一线程中按序处理命令和响应
//! 4. 大幅降低内存开销，提高性能

use std::collections::{HashMap, VecDeque};
// use std::sync::{Arc, Mutex}; // 暂时未使用
use std::thread;
use std::time::Duration;

use crate::protocol::{NetworkStream, RespParser};
use crate::storage::{Command, CommandResult, SingleThreadCommandDispatcher};
use crate::event::ConnectionId;

/// 客户端连接状态（类似Redis的client结构）
#[derive(Debug)]
pub struct ClientState {
    /// 网络流
    pub network_stream: Box<dyn NetworkStream>,
    /// 输入缓冲区（接收数据）
    pub input_buffer: Vec<u8>,
    /// 输出队列（待发送的响应，严格按序）
    pub output_queue: VecDeque<Vec<u8>>,
    /// RESP协议解析器
    pub resp_parser: RespParser,
    /// 待处理命令队列
    pub pending_commands: VecDeque<Box<dyn Command>>,
    /// 连接ID
    pub connection_id: ConnectionId,
    /// 是否活跃
    pub is_active: bool,
    /// 最后活跃时间
    pub last_active: std::time::Instant,
}

impl ClientState {
    /// 创建新的客户端状态
    pub fn new(connection_id: ConnectionId, network_stream: Box<dyn NetworkStream>) -> Self {
        Self {
            network_stream,
            input_buffer: Vec::with_capacity(16384), // 16KB初始缓冲区
            output_queue: VecDeque::new(),
            resp_parser: RespParser::new(),
            pending_commands: VecDeque::new(),
            connection_id,
            is_active: true,
            last_active: std::time::Instant::now(),
        }
    }

    /// 读取数据到输入缓冲区
    pub fn read_data(&mut self) -> Result<usize, String> {
        let mut temp_buffer = vec![0u8; 8192];
        
        match self.network_stream.read(&mut temp_buffer) {
            Ok(0) => {
                // 连接关闭
                self.is_active = false;
                Ok(0)
            }
            Ok(n) => {
                self.input_buffer.extend_from_slice(&temp_buffer[..n]);
                self.last_active = std::time::Instant::now();
                Ok(n)
            }
            Err(e) => {
                self.is_active = false;
                Err(format!("读取数据失败: {e}"))
            }
        }
    }

    /// 解析命令
    pub fn parse_commands(&mut self) -> Result<usize, String> {
        self.resp_parser.feed_data(&self.input_buffer);
        self.input_buffer.clear(); // 数据已交给解析器

        let mut parsed_count = 0;

        loop {
            match self.resp_parser.parse_value() {
                Ok(Some(resp_value)) => {
                    // 成功解析出一个RESP值，转换为命令
                    match self.resp_parser.resp_to_command(resp_value) {
                        Ok(command) => {
                            self.pending_commands.push_back(command);
                            parsed_count += 1;
                        }
                        Err(e) => {
                            // 协议错误，添加错误响应到输出队列
                            self.output_queue.push_back(b"-ERR Protocol error\r\n".to_vec());
                            eprintln!("❌ 连接 {} 命令转换失败: {:?}", self.connection_id, e);
                        }
                    }
                }
                Ok(None) => {
                    // 数据不完整，等待更多数据
                    break;
                }
                Err(e) => {
                    // 解析错误
                    self.output_queue.push_back(b"-ERR Protocol error\r\n".to_vec());
                    eprintln!("❌ 连接 {} RESP解析错误: {:?}", self.connection_id, e);
                    break;
                }
            }
        }

        Ok(parsed_count)
    }

    /// 添加响应到输出队列
    pub fn add_response(&mut self, result: CommandResult) {
        let resp_bytes = result.to_resp().into_bytes();
        self.output_queue.push_back(resp_bytes);
    }

    /// 发送输出队列中的响应
    pub fn flush_responses(&mut self) -> Result<usize, String> {
        let mut sent_count = 0;

        while let Some(response) = self.output_queue.pop_front() {
            match self.network_stream.write(&response) {
                Ok(_) => {
                    sent_count += 1;
                    self.last_active = std::time::Instant::now();
                }
                Err(e) => {
                    // 写入失败，重新放回队列头部
                    self.output_queue.push_front(response);
                    return Err(format!("写入响应失败: {e}"));
                }
            }
        }

        if sent_count > 0 {
            let _ = self.network_stream.flush();
        }

        Ok(sent_count)
    }

    /// 检查连接是否超时
    pub fn is_timeout(&self, timeout: Duration) -> bool {
        self.last_active.elapsed() > timeout
    }
}

/// Redis风格的网络服务器
/// 
/// 核心设计：
/// 1. 单线程事件循环（类似Redis主线程）
/// 2. 每连接独立缓冲区（类似Redis client结构）
/// 3. 同步命令处理（保证顺序）
/// 4. 大幅降低内存开销
pub struct RedisStyleServer {
    /// 所有客户端连接状态
    clients: HashMap<ConnectionId, ClientState>,
    /// 存储命令分发器
    storage_dispatcher: SingleThreadCommandDispatcher,
    /// 连接ID计数器
    connection_counter: u64,
    /// 是否运行中
    running: bool,
    /// 配置
    config: RedisStyleConfig,
}

/// 服务器配置
#[derive(Debug, Clone)]
pub struct RedisStyleConfig {
    /// 连接超时时间
    pub connection_timeout: Duration,
    /// 最大连接数
    pub max_connections: usize,
    /// 事件循环间隔
    pub loop_interval: Duration,
}

impl Default for RedisStyleConfig {
    fn default() -> Self {
        Self {
            connection_timeout: Duration::from_secs(300), // 5分钟
            max_connections: 10000,
            loop_interval: Duration::from_millis(1), // 1毫秒
        }
    }
}

impl RedisStyleServer {
    /// 创建新的Redis风格服务器
    pub fn new(storage_dispatcher: SingleThreadCommandDispatcher, config: Option<RedisStyleConfig>) -> Self {
        Self {
            clients: HashMap::new(),
            storage_dispatcher,
            connection_counter: 0,
            running: false,
            config: config.unwrap_or_default(),
        }
    }

    /// 启动服务器
    pub fn start(&mut self) -> Result<(), String> {
        if self.running {
            return Err("服务器已在运行".to_string());
        }

        self.running = true;
        self.storage_dispatcher.start();
        
        println!("🚀 Redis风格服务器已启动");
        Ok(())
    }

    /// 停止服务器
    pub fn stop(&mut self) -> Result<(), String> {
        if !self.running {
            return Ok(());
        }

        self.running = false;
        self.storage_dispatcher.stop();
        self.clients.clear();
        
        println!("🛑 Redis风格服务器已停止");
        Ok(())
    }

    /// 接受新连接
    pub fn accept_connection(&mut self, network_stream: Box<dyn NetworkStream>) -> Result<ConnectionId, String> {
        if self.clients.len() >= self.config.max_connections {
            return Err(format!("连接数已达上限: {}", self.config.max_connections));
        }

        self.connection_counter += 1;
        let connection_id = self.connection_counter;

        let client_state = ClientState::new(connection_id, network_stream);
        self.clients.insert(connection_id, client_state);

        println!("🌐 接受新连接: ID={connection_id}");
        Ok(connection_id)
    }

    /// 主事件循环（类似Redis的aeMain）
    pub fn event_loop(&mut self) -> Result<(), String> {
        while self.running {
            // 1. 处理所有连接的读事件
            self.process_read_events()?;

            // 2. 处理所有待执行命令
            self.process_commands()?;

            // 3. 处理所有连接的写事件
            self.process_write_events()?;

            // 4. 清理超时连接
            self.cleanup_timeout_connections();

            // 5. 短暂休眠，避免CPU空转
            thread::sleep(self.config.loop_interval);
        }

        Ok(())
    }

    /// 处理所有连接的读事件
    fn process_read_events(&mut self) -> Result<(), String> {
        let mut to_remove = Vec::new();

        for (conn_id, client) in self.clients.iter_mut() {
            if !client.is_active {
                to_remove.push(*conn_id);
                continue;
            }

            // 读取数据
            match client.read_data() {
                Ok(0) => {
                    // 连接关闭
                    to_remove.push(*conn_id);
                }
                Ok(n) => {
                    if n > 0 {
                        // 解析命令
                        if let Err(e) = client.parse_commands() {
                            eprintln!("❌ 连接 {conn_id} 解析命令失败: {e}");
                        }
                    }
                }
                Err(e) => {
                    eprintln!("❌ 连接 {conn_id} 读取失败: {e}");
                    to_remove.push(*conn_id);
                }
            }
        }

        // 移除断开的连接
        for conn_id in to_remove {
            self.clients.remove(&conn_id);
            println!("🗑️ 移除连接: {conn_id}");
        }

        Ok(())
    }

    /// 处理所有待执行命令
    fn process_commands(&mut self) -> Result<(), String> {
        // 收集所有待处理的命令
        let mut commands_to_process = Vec::new();

        for (conn_id, client) in self.clients.iter_mut() {
            while let Some(command) = client.pending_commands.pop_front() {
                commands_to_process.push((*conn_id, command));
            }
        }

        // 处理命令（在同一线程中，保证顺序）
        for (conn_id, command) in commands_to_process {
            let result = match self.storage_dispatcher.dispatch_command(command) {
                Ok(()) => {
                    // 简化：模拟命令执行结果
                    CommandResult::Ok // 实际应该从存储层获取结果
                }
                Err(e) => {
                    CommandResult::Error(format!("执行失败: {e}"))
                }
            };

            // 将结果添加到对应连接的输出队列
            if let Some(client) = self.clients.get_mut(&conn_id) {
                client.add_response(result);
            }
        }

        Ok(())
    }

    /// 处理所有连接的写事件
    fn process_write_events(&mut self) -> Result<(), String> {
        for (conn_id, client) in self.clients.iter_mut() {
            if let Err(e) = client.flush_responses() {
                eprintln!("❌ 连接 {conn_id} 发送响应失败: {e}");
            }
        }

        Ok(())
    }

    /// 清理超时连接
    fn cleanup_timeout_connections(&mut self) {
        let mut to_remove = Vec::new();

        for (conn_id, client) in self.clients.iter() {
            if client.is_timeout(self.config.connection_timeout) {
                to_remove.push(*conn_id);
            }
        }

        for conn_id in to_remove {
            self.clients.remove(&conn_id);
            println!("🧹 清理超时连接: {conn_id}");
        }
    }

    /// 获取统计信息
    pub fn get_stats(&self) -> RedisStyleStats {
        RedisStyleStats {
            active_connections: self.clients.len(),
            total_input_buffer_size: self.clients.values()
                .map(|c| c.input_buffer.len())
                .sum(),
            total_output_queue_size: self.clients.values()
                .map(|c| c.output_queue.len())
                .sum(),
            total_pending_commands: self.clients.values()
                .map(|c| c.pending_commands.len())
                .sum(),
        }
    }
}

/// 服务器统计信息
#[derive(Debug, Clone)]
pub struct RedisStyleStats {
    pub active_connections: usize,
    pub total_input_buffer_size: usize,
    pub total_output_queue_size: usize,
    pub total_pending_commands: usize,
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io;

    #[derive(Debug)]
    struct MockStream {
        data: Vec<u8>,
        pos: usize,
        responses: Vec<u8>,
    }

    impl MockStream {
        fn new(data: Vec<u8>) -> Self {
            Self { data, pos: 0, responses: Vec::new() }
        }
    }

    impl crate::protocol::NetworkStream for MockStream {
        fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
            if self.pos >= self.data.len() { return Ok(0); }
            let to_copy = std::cmp::min(buf.len(), self.data.len() - self.pos);
            buf[..to_copy].copy_from_slice(&self.data[self.pos..self.pos + to_copy]);
            self.pos += to_copy;
            Ok(to_copy)
        }
        
        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
            self.responses.extend_from_slice(buf);
            Ok(buf.len())
        }
        
        fn flush(&mut self) -> io::Result<()> { Ok(()) }
        fn is_connected(&self) -> bool { true }
        fn remote_addr(&self) -> Option<String> { Some("test".to_string()) }
        fn shutdown(&mut self) -> io::Result<()> { Ok(()) }
    }

    #[test]
    fn test_redis_style_server() {
        let mut server = RedisStyleServer::new(
            SingleThreadCommandDispatcher::new(),
            None
        );
        
        server.start().unwrap();
        
        // 接受连接
        let stream = Box::new(MockStream::new(b"*2\r\n$4\r\nPING\r\n".to_vec()));
        let conn_id = server.accept_connection(stream).unwrap();
        
        assert_eq!(server.clients.len(), 1);
        
        server.stop().unwrap();
    }
}